from pyflink.common import Duration, WatermarkStrategy
from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode
from pyflink.datastream.connectors.file_system import FileSource, StreamFormat

# 1、创建flink执行环境
env = StreamExecutionEnvironment.get_execution_environment()

env.set_runtime_mode(RuntimeExecutionMode.STREAMING)

file_source = FileSource \
    .for_record_stream_format(StreamFormat.text_line_format(), "hdfs://master:9000/data/students") \
    .monitor_continuously(Duration.of_seconds(1)) \
    .build()

# 基于file source构建DS(无界流)
lines_ds = env.from_source(file_source, WatermarkStrategy.no_watermarks(), "FileSource")

# sum: 按照下标求和
clazz_num_ds = lines_ds \
    .map(lambda line: (line.split(",")[-1], 1)) \
    .key_by(lambda kv: kv[0]) \
    .sum(1)

clazz_num_ds.print()

env.execute()
